package com.wande.dataplatform.service.impl;

import cn.hutool.core.util.StrUtil;
import com.wande.dataplatform.common.constant.DataPlatformConstants;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.common.exception.DataPlatformException;
import com.wande.dataplatform.common.utils.DorisConnectionManager;
import com.wande.dataplatform.config.DorisProperties;
import com.wande.dataplatform.service.IDorisQueryService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.util.*;

/**
 * Doris 查询服务实现
 * 
 * @author ruoyi
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DorisQueryServiceImpl implements IDorisQueryService {

    private final DorisConnectionManager connectionManager;
    private final DorisProperties dorisProperties;

    @Override
    public List<Map<String, Object>> executeQuery(String sql) {
        return executeQuery(sql, dorisProperties.getQueryTimeout());
    }

    @Override
    public List<Map<String, Object>> executeQuery(String sql, int timeoutSeconds) {
        if (StrUtil.isBlank(sql)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL 语句不能为空");
        }

        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;
        List<Map<String, Object>> results = new ArrayList<>();

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            stmt.setQueryTimeout(timeoutSeconds);

            log.debug("执行查询 SQL: {}", sql);
            long startTime = System.currentTimeMillis();

            rs = stmt.executeQuery(sql);
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();

            while (rs.next()) {
                Map<String, Object> row = new LinkedHashMap<>();
                for (int i = 1; i <= columnCount; i++) {
                    String columnName = metaData.getColumnLabel(i);
                    Object value = rs.getObject(i);
                    row.put(columnName, value);
                }
                results.add(row);
            }

            long endTime = System.currentTimeMillis();
            log.info("查询完成，返回 {} 条记录，耗时 {} ms", results.size(), endTime - startTime);

        } catch (SQLTimeoutException e) {
            log.error("查询超时: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "查询超时");
        } catch (SQLException e) {
            log.error("查询失败: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "查询失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return results;
    }

    @Override
    public Map<String, Object> executePageQuery(String sql, int pageNum, int pageSize) {
        return executePageQuery(sql, pageNum, pageSize, dorisProperties.getQueryTimeout());
    }

    @Override
    public Map<String, Object> executePageQuery(String sql, int pageNum, int pageSize, int timeoutSeconds) {
        if (pageNum < 1) {
            pageNum = 1;
        }
        if (pageSize < 1 || pageSize > DataPlatformConstants.Default.PAGE_SIZE) {
            pageSize = DataPlatformConstants.Default.PAGE_SIZE;
        }

        // 查询总记录数
        String countSql = buildCountSql(sql);
        long total = queryForCount(countSql);

        // 构建分页 SQL
        int offset = (pageNum - 1) * pageSize;
        String pageSql = sql + " LIMIT " + offset + ", " + pageSize;

        // 执行分页查询
        List<Map<String, Object>> records = executeQuery(pageSql, timeoutSeconds);

        // 构建分页结果
        Map<String, Object> result = new HashMap<>();
        result.put("records", records);
        result.put("total", total);
        result.put("pageNum", pageNum);
        result.put("pageSize", pageSize);
        result.put("pages", (total + pageSize - 1) / pageSize);

        return result;
    }

    @Override
    public int executeUpdate(String sql) {
        return executeUpdate(sql, dorisProperties.getQueryTimeout());
    }

    @Override
    public int executeUpdate(String sql, int timeoutSeconds) {
        if (StrUtil.isBlank(sql)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL 语句不能为空");
        }

        Connection conn = null;
        Statement stmt = null;
        int affectedRows = 0;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            stmt.setQueryTimeout(timeoutSeconds);

            log.debug("执行更新 SQL: {}", sql);
            long startTime = System.currentTimeMillis();

            affectedRows = stmt.executeUpdate(sql);

            long endTime = System.currentTimeMillis();
            log.info("更新完成，影响 {} 行，耗时 {} ms", affectedRows, endTime - startTime);

        } catch (SQLTimeoutException e) {
            log.error("更新超时: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "更新超时");
        } catch (SQLException e) {
            log.error("更新失败: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "更新失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(null, stmt, conn);
        }

        return affectedRows;
    }

    @Override
    public int[] executeBatch(List<String> sqlList) {
        if (sqlList == null || sqlList.isEmpty()) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL 列表不能为空");
        }

        Connection conn = null;
        Statement stmt = null;
        int[] results = null;

        try {
            conn = connectionManager.getConnection();
            conn.setAutoCommit(false);
            stmt = conn.createStatement();

            log.debug("批量执行 {} 条 SQL", sqlList.size());
            long startTime = System.currentTimeMillis();

            for (String sql : sqlList) {
                if (StrUtil.isNotBlank(sql)) {
                    stmt.addBatch(sql);
                }
            }

            results = stmt.executeBatch();
            conn.commit();

            long endTime = System.currentTimeMillis();
            log.info("批量执行完成，耗时 {} ms", endTime - startTime);

        } catch (SQLException e) {
            log.error("批量执行失败", e);
            if (conn != null) {
                try {
                    conn.rollback();
                } catch (SQLException ex) {
                    log.error("回滚失败", ex);
                }
            }
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "批量执行失败: " + e.getMessage());
        } finally {
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                } catch (SQLException e) {
                    log.warn("恢复自动提交失败", e);
                }
            }
            connectionManager.closeResources(null, stmt, conn);
        }

        return results;
    }

    @Override
    public Object queryForObject(String sql) {
        if (StrUtil.isBlank(sql)) {
            throw new DataPlatformException(DataPlatformErrorCode.PARAM_INVALID, "SQL 语句不能为空");
        }

        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;
        Object result = null;

        try {
            conn = connectionManager.getConnection();
            stmt = conn.createStatement();
            stmt.setQueryTimeout(dorisProperties.getQueryTimeout());

            rs = stmt.executeQuery(sql);

            if (rs.next()) {
                result = rs.getObject(1);
            }

        } catch (SQLException e) {
            log.error("查询单个值失败: {}", sql, e);
            throw new DataPlatformException(DataPlatformErrorCode.DORIS_QUERY_FAILED, "查询失败: " + e.getMessage());
        } finally {
            connectionManager.closeResources(rs, stmt, conn);
        }

        return result;
    }

    @Override
    public long queryForCount(String sql) {
        Object result = queryForObject(sql);
        if (result == null) {
            return 0L;
        }

        if (result instanceof Number) {
            return ((Number) result).longValue();
        }

        try {
            return Long.parseLong(result.toString());
        } catch (NumberFormatException e) {
            log.warn("无法将查询结果转换为数字: {}", result);
            return 0L;
        }
    }

    /**
     * 构建 COUNT SQL
     * 
     * @param sql 原始 SQL
     * @return COUNT SQL
     */
    private String buildCountSql(String sql) {
        // 简单实现：将 SELECT ... FROM 替换为 SELECT COUNT(*) FROM
        String upperSql = sql.toUpperCase().trim();
        
        if (upperSql.startsWith("SELECT")) {
            int fromIndex = upperSql.indexOf("FROM");
            if (fromIndex > 0) {
                String fromPart = sql.substring(fromIndex);
                
                // 移除 ORDER BY 子句
                int orderByIndex = fromPart.toUpperCase().lastIndexOf("ORDER BY");
                if (orderByIndex > 0) {
                    fromPart = fromPart.substring(0, orderByIndex);
                }
                
                // 移除 LIMIT 子句
                int limitIndex = fromPart.toUpperCase().lastIndexOf("LIMIT");
                if (limitIndex > 0) {
                    fromPart = fromPart.substring(0, limitIndex);
                }
                
                return "SELECT COUNT(*) " + fromPart.trim();
            }
        }
        
        // 如果无法解析，返回原 SQL 包装
        return "SELECT COUNT(*) FROM (" + sql + ") AS count_table";
    }

}
